package org.orz.cloud.lock.zookeeper;

import org.apache.curator.framework.CuratorFramework;
import org.apache.curator.framework.recipes.locks.Revoker;
import org.apache.zookeeper.data.Stat;
import org.orz.cloud.lock.DistributedLock;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;

import java.util.concurrent.TimeUnit;

/**
 * Zookeeper锁的实际对象
 */
public class ZkLock implements DistributedLock {
	public static final Logger LOGGER = LoggerFactory.getLogger(ZkLock.class);
	
	private static final int DEFAULT_TIMEOUT = 5000;

	private static CuratorFramework curator;
	
	private static String PREFIX = "/DLOCK/";
	
	private static String TEMP_KEY_PATH = PREFIX+"CURRENTKEY";
	
	private ZkProcessMutex lock;
	
	private String lockPath;
	
	private String lockKey;

	/**
	 * 锁的地址，这个应该是需要防止其他线程或者其他服务释放当前锁的一个操作，
	 * 但是目前看起来是没有其效果，试一下
	 */
	private String lockTmpKey;

	@Override
	public boolean lock(String key){
		return lock(key,DEFAULT_TIMEOUT);
	}

	@Override
	public boolean lock(String key, long timeout){
		try {
			lockKey = PREFIX+key;
			lockTmpKey = TEMP_KEY_PATH+"/"+key;
			//如果在zk服务器中，之前存在死锁，就释放掉
			clean(false);
			//创建锁对象
			createLock(key);
			if(lock.acquire(timeout,TimeUnit.MILLISECONDS)){
				lockPath = lock.getZkLockPath();
				//在zk中保存当前的锁对象
				saveCurrentLockKey();
				if(LOGGER.isInfoEnabled())
					LOGGER.info("Get the lock:"+lockKey+",lockPath:"+lockPath);
				return true;
			}else{
				//如果在zk服务器中存在死锁，就释放掉，同时删除当前的锁
				clean(true);
				return false;
			}
		} catch (Exception e) {
			if(LOGGER.isWarnEnabled())
				LOGGER.warn("DistributedLock acquire error",e);
			return false;
		}
	}

	@Override
	public void release() {
		try {
			if(lock != null){
				lock.release();
			}
        } catch (Exception e) {
        	if(LOGGER.isWarnEnabled())
        		LOGGER.warn("DistributedLock release error",e);
        }finally{
			delete(lockKey);
			delete(lockTmpKey);
        }
	}

	/**
	 * 设置zk锁客户端对象
	 * @param curatorP 客户端锁对象
	 */
	static void setCurator(CuratorFramework curatorP) {
		curator = curatorP;
		Stat stat;
		try {
			stat = curator.checkExists().forPath(TEMP_KEY_PATH);
			if(stat == null){
				curator.create().creatingParentsIfNeeded().forPath(TEMP_KEY_PATH);
			}
		} catch (Exception e) {
			if(LOGGER.isInfoEnabled())
				LOGGER.info("Create tmp node error!",e);
			throw new RuntimeException(e);
		}
	}

	/**
	 * 创建锁，这里的 {@code new ZKProcessMutex}是真正的创建锁的操作
	 * @param key 锁的key值
	 */
	private void createLock(String key){
		lock = new ZkProcessMutex(curator, PREFIX+key);
		//这里是设置锁可以被其他线程或其他远程线程释放。但是实现里面好像又没有撤销。这个不是很懂，随后验证一下
		lock.makeRevocable(paramT -> {
			if(paramT.isAcquiredInThisProcess()){
				if(LOGGER.isWarnEnabled())
					LOGGER.warn("The lock:"+lockKey+",lockPath:"+lockPath+" is unlocked by other thread!");
			}else{
				if(LOGGER.isWarnEnabled())
					LOGGER.warn("The lock:"+lockKey+",lockPath:"+lockPath+" is unlocked by other remote thread!");
			}
		});
	}

	/**
	 * 清除锁，包括释放死锁（当前的key最后一次修改的时间超过30秒就算是死锁），
	 * 如果要删除当前锁，也会删除当前锁
	 * @param isDelete 是否删除当前的锁
	 */
	private void clean(boolean isDelete) throws Exception{
		Stat stat = curator.checkExists().forPath(lockTmpKey);
		if(stat != null && System.currentTimeMillis() - stat.getMtime()>30000){
			releaseDeadLock();
			if(isDelete){
				delete(lockKey);
				delete(lockTmpKey);
			}
		}
	}

	/**
	 * 释放死锁
	 */
	private void releaseDeadLock(){
		if(LOGGER.isWarnEnabled())
			LOGGER.warn("Start release the time out lock:"+lockKey+",lockPath:"+lockPath);
		
		String delkey = "";
		try {
			delkey = new String(curator.getData().forPath(lockTmpKey));
			if(delkey.startsWith("/")){
				Revoker.attemptRevoke(curator, delkey);
				deleteWithCheck(delkey);
				if(LOGGER.isWarnEnabled())
					LOGGER.warn("End release the time out lock:"+lockKey+",lockPath:"+delkey);
			}else{
				if(LOGGER.isWarnEnabled())
					LOGGER.warn("Error release the time out lock:"+lockKey+",lockPath:"+delkey);
			}
		} catch (Exception e) {
			if(LOGGER.isErrorEnabled())
				LOGGER.error("Error release the time out lock:"+lockKey+",lockPath:"+delkey,e);
		}
	}

	/**
	 * 删除锁，并在删除之前检查是否存在
	 * @param key 锁的key
	 */
	private void deleteWithCheck(String key){
		Stat delStat;
		try {
			delStat = curator.checkExists().forPath(key);
		} catch (Exception e) {
			if(LOGGER.isErrorEnabled())
				LOGGER.error("Error clean the key:"+key,e);
			return;
		}
		if(delStat != null){
			delete(key);
		}
	}

	/**
	 * 删除锁
	 * @param key 锁的key
	 */
	private void delete(String key){
		try {
			curator.delete().deletingChildrenIfNeeded().forPath(key);
		} catch (Exception e) {
			if(LOGGER.isErrorEnabled())
				LOGGER.error("Error clean the key:"+key,e);
		}
	}

	/**
	 * 在zk中保存当前创建的锁对象
	 */
	private void saveCurrentLockKey(){
		Stat stat;
		try {
			stat = curator.checkExists().forPath(lockTmpKey);
			if( stat != null ){
				curator.setData().forPath(lockTmpKey, lockPath.getBytes());
			}else{
				curator.create().forPath(lockTmpKey, lockPath.getBytes());
			}
		} catch (Exception e) {
			if(LOGGER.isErrorEnabled())
				LOGGER.error("Error save the key:"+lockTmpKey+",lockPath:"+lockPath,e);
		}
	}
}
